package storm;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * @author Hanhan_Jun
 * @date 2021年12月08日 9:48
 */
public class Topology {
    public static void main(String[] args) throws AuthorizationException, InvalidTopologyException, AlreadyAliveException {

        TopologyBuilder builder = new TopologyBuilder();
        //设置Spout1并行度为3
        builder.setSpout("Spout1", new Spout1(),3);
        //设置Bolt的线程数为10，字段分组策略，设置task数量为6
        builder.setBolt("Bolt1_1",new Bolt1_1(),10).fieldsGrouping("Spout",new Fields("book_name", "book_num", "book_press","book_type","book_price")).setNumTasks(6);
        //设置Spout2并行度为3
        builder.setSpout("Spout2",new Spout2(),3);
        //设置Bolt的线程数为10，字段分组策略，设置task数量为6
        builder.setBolt("Bolt2_1",new Bolt2_1(),4).fieldsGrouping("Spout2",new Fields("recipient", "receiveplace", "total","methodofpayment","paytime"));

        //提交集群运行，注释掉本地运行代码，打包的时候同依赖包一起打包
        Config config = new Config();
        StormSubmitter.submitTopology("task", config, builder.createTopology());


        //本地运行
//        LocalCluster localCluster = new LocalCluster();
//        Config config = new Config();
//        localCluster.submitTopology("topology1", config,
//                builder.createTopology());
    }
}
